package com.bdqn.mr.partition;

import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Partitioner;

/**
 * 第一步：自己写一个普通的Java类
 * 第二步：要想成为分区器，继承Partitioner抽象类
 */
public class PhonePartitioner extends Partitioner<Text, FlowBean> {


    /**
     * 根据 map 写出的 KV 键值对，获取 KV 键值对要进入的分区编号
     *
     * @param text          map输出的 key
     * @param flowBean      map输出的 value
     * @param numPartitions 分区总数 = ReduceTask的数量（在Driver中设置）
     * @return
     */
    @Override
    public int getPartition(Text text, FlowBean flowBean, int numPartitions) {
        // 136 -> 分区0
        // 137 -> 分区1
        // 138 -> 分区2
        // 139 -> 分区3
        // 其他 -> 分区4
        // 13622445566
        String phone = text.toString();

        if (phone.startsWith("136")) {
            return 0;
        } else if (phone.startsWith("137")) {
            return 1;
        } else if (phone.startsWith("138")) {
            return 2;
        } else if (phone.startsWith("139")) {
            return 3;
        } else {
            return 4;
        }
    }
}
